package ai.people.netmon.warning.service.impl;

import ai.people.core.common.util.ApplicationContextUtil;
import ai.people.core.grpc.lib.dict.DictRequest;
import ai.people.core.grpc.lib.dict.DictResponse;
import ai.people.core.grpc.lib.dict.DictResponseData;
import ai.people.core.grpc.lib.dict.DictServiceGrpc;
import ai.people.core.grpc.lib.es.*;
import ai.people.core.grpc.utils.ProtoJsonUtil;
import ai.people.core.security.support.RequestHolderUtil;
import ai.people.netmon.framework.domain.warning.entity.*;
import ai.people.netmon.framework.domain.warning.po.WarningAccountAnalysisPO;
import ai.people.netmon.framework.domain.warning.request.account.AccountJoinedListRequest;
import ai.people.netmon.framework.domain.warning.request.account.AccountWarningInfoRequest;
import ai.people.netmon.framework.domain.warning.request.account.WarningAccountAnalysisParam;
import ai.people.netmon.framework.domain.warning.request.account.WarningAccountRequest;
import ai.people.netmon.framework.domain.warning.vo.*;
import ai.people.netmon.framework.exception.enums.AccountExceptionEnum;
import ai.people.netmon.framework.exception.enums.CommonEnum;
import ai.people.netmon.framework.model.response.ListResponse;
import ai.people.netmon.framework.model.response.PageResponse;
import ai.people.netmon.framework.utils.AssertUtils;
import ai.people.netmon.utils.StringPool;
import ai.people.netmon.utils.StringUtils;
import ai.people.netmon.utils.localdateUtil.DateTimeFormatterUtil;
import ai.people.netmon.warning.dict.DictGrpcClient;
import ai.people.netmon.warning.mapper.WarningAccountMapper;
import ai.people.netmon.warning.service.*;
import com.alibaba.fastjson.JSON;
import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import com.google.common.base.Strings;
import com.google.protobuf.Message;
import lombok.RequiredArgsConstructor;
import net.devh.boot.grpc.client.inject.GrpcClient;
import org.redisson.api.RLock;
import org.redisson.api.RedissonClient;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;
import org.springframework.util.CollectionUtils;

import java.util.*;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.function.Function;
import java.util.stream.Collectors;
import java.util.stream.Stream;

/**
 * @author ligua
 * @description 针对表【warning_account(预警账号表)】的数据库操作Service实现
 * @createDate 2022-07-28 17:50:38
 */
@Service
@RequiredArgsConstructor
public class WarningAccountServiceImpl extends ServiceImpl<WarningAccountMapper, WarningAccount>
        implements WarningAccountService, WarningSpeechTemplateService<AccountWarningInfoRequest, MessageContentVO> {

    private final RedissonClient redissonClient;

    private final WarningAccountMapper warningAccountMapper;

    private final WarningAccountLabelService warningAccountLabelServiceImpl;

    @Autowired
    private WarningAccountAnalysisService warningAccountAnalysisServiceImpl;

    private final WarningAccountCollectService warningAccountCollectServiceImpl;

    private final WarningLabelTreeService warningLabelTreeServiceImpl;

    //private final WarningAccountPushGrpcClient warningAccountPushGrpcClient;

    private final DictGrpcClient dictGrpcClient;

    @GrpcClient(value = "NM-SERVICE-SEARCH-MSG")
    private BaseSearchServiceGrpc.BaseSearchServiceBlockingStub searchServiceBlockingStub;

    @GrpcClient(value = "NM-SERVICE-SEARCH-MSG")
    private SearchAccountServiceGrpc.SearchAccountServiceBlockingStub searchAccountServiceBlockingStub;

    @GrpcClient(value = "NM-SERVICE-SYSTEM")
    private DictServiceGrpc.DictServiceBlockingStub dictServiceBlockingStub;

    private final WarningDataSyncNotifyService warningDataSyncNotifyService;

    /**
     * 预警账号的网络活动信息
     *
     * @param accountWarningRequest@return {@link PageResponse}<{@link ?}>
     * @return {@link PageResponse}<{@link MessageContentVO}>
     */
    @Override
    public PageResponse<MessageContentVO> searchQuery(AccountWarningInfoRequest accountWarningRequest) {
        // 查询账号配置的查询条件
        WarningAccountVO accountBaseInfo = this.getAccountBaseInfo(accountWarningRequest.getAccountId());
        AssertUtils.isNotNull(accountBaseInfo, AccountExceptionEnum.ACCOUNT_DOES_NOT_EXIST);
        String account = accountBaseInfo.getAccount();
        // 封装ES查询条件
        SearchRequest.Builder builder = SearchRequest.newBuilder().setSource(accountBaseInfo.getPlatform()).setSort("-ps").setLimit(accountWarningRequest.getPageSize()).setOffset(accountWarningRequest.getPageNo());
        if (StringUtils.isNotBlank(account)) {
            builder.setAccountId(account).setAccountName(account);
        }
        if (StringUtils.isNotBlank(accountBaseInfo.getAccountId())) {
            builder.setAccountId(accountBaseInfo.getAccountId()).setAccountName(accountBaseInfo.getAccountId());
        }
        if (Objects.nonNull(accountWarningRequest.getStartTime())) {
            builder.setStartDate(DateTimeFormatterUtil.localDateTimeFormatterToDateTime(accountWarningRequest.getStartTime()));
        }
        if (Objects.nonNull(accountWarningRequest.getEndTime())) {
            builder.setEndDate(DateTimeFormatterUtil.localDateTimeFormatterToDateTime(accountWarningRequest.getEndTime()));
        }
        // 情感标签
        builder.setSentiment(Strings.nullToEmpty(accountWarningRequest.getEmotion()));
        // 业务标签
        builder.setTags(Strings.nullToEmpty(accountWarningRequest.getLabel()));
        SearchPageResponse searchPageResponse = searchServiceBlockingStub.baseSearch(builder.build());
        Map<String, WarningLabelTree> warningLabelTreeMap = warningLabelTreeServiceImpl.allLabelValueMapping();
        // 处理检索结果
        List<MessageContentVO> messageContentVOList = searchPageResponse.getListList().stream()
                .map(searchResponseData -> this.objectConversion(searchResponseData, dictGrpcClient.getDictMap("warning_message_emotion"), warningLabelTreeMap))
                .collect(Collectors.toList());
        return PageResponse.builder(messageContentVOList, searchPageResponse.getTotal());
    }


    /**
     * 重点账号导航页
     *
     * @param userId 用户id
     * @return {@link List}<{@link AccountNavigationVO}>
     */
    @Override
    public List<AccountNavigationVO> keyAccountNavigation(Long userId) {
        // 查询此用户关注的账号
        List<WarningAccount> warningAccounts = warningAccountMapper.listJoinAnalysisAccount(RequestHolderUtil.getUserId(), new AccountJoinedListRequest());
        if (CollectionUtils.isEmpty(warningAccounts)) {
            return this.accountNavigationTree(Collections.emptyMap());
        }
        // 提取账号
        String accounts = warningAccounts.stream().map(WarningAccount::getAccount).collect(Collectors.joining(StringPool.COMMA));
        // 提取账号平台
        String platform = warningAccounts.stream().map(WarningAccount::getPlatform).distinct().collect(Collectors.joining(StringPool.COMMA));
        platform = Stream.of(platform.split(StringPool.COMMA)).distinct().collect(Collectors.joining(StringPool.COMMA));
        // ES查询这些账号的发言被打标签的信息
        AccountWarningSpeechGrpcRequest speechGrpcRequest = AccountWarningSpeechGrpcRequest.newBuilder().setPlatform(platform).setAccount(accounts).build();
        AccountWarningSpeechGrpcResponse accountWarningSpeech = searchServiceBlockingStub.queryAccountWarningSpeech(speechGrpcRequest);
        Map<String, Integer> accountMessageCount = accountWarningSpeech.getListList().stream().collect(Collectors.toMap(AccountWarningSpeechGrpcData::getAccount, AccountWarningSpeechGrpcData::getMessageCount));
        Map<String, List<String>> labelAccountMap = new HashMap<>();

        List<Long> accountIds = warningAccounts.stream().map(WarningAccount::getId).collect(Collectors.toList());
        // 查询账号的标签
        List<WarningAccountLabelRelation> warningAccountLabelRelations = warningLabelTreeServiceImpl.queryLabelByAccount(accountIds);
        Map<Long, List<WarningAccountLabelRelation>> accountLabelMap = warningAccountLabelRelations.stream().collect(Collectors.groupingBy(WarningAccountLabelRelation::getWarningAccountId));
        warningAccounts.forEach(warningAccount -> {
            // 统计标签下的账号数量
            for (WarningAccountLabelRelation warningAccountLabelRelation : accountLabelMap.getOrDefault(warningAccount.getId(), Collections.emptyList())) {
                List<String> accountMapOrDefault = labelAccountMap.getOrDefault(warningAccountLabelRelation.getLabelValue(), new ArrayList<>());
                accountMapOrDefault.add(warningAccount.getAccount());
                labelAccountMap.put(warningAccountLabelRelation.getLabelValue(), accountMapOrDefault);
            }
        });
        List<AccountNavigationVO> accountNavigationVOS = new ArrayList<>();
        labelAccountMap.forEach((labelValue, accountCounts) -> {
            AccountNavigationVO accountNavigationVO = new AccountNavigationVO();
            accountNavigationVO.setLabelValue(labelValue);
            accountNavigationVO.setAccountCount(accountCounts.size());
            Integer messageCount = accountCounts.stream().map(account -> accountMessageCount.getOrDefault(account, 0)).reduce(Integer::sum).orElse(0);
            accountNavigationVO.setMessageCount(messageCount);
            accountNavigationVOS.add(accountNavigationVO);
        });

        Map<String, AccountNavigationVO> navigationVOMap = accountNavigationVOS.stream().collect(Collectors.toMap(AccountNavigationVO::getLabelValue, Function.identity()));
        return this.accountNavigationTree(navigationVOMap);
    }


    /**
     * 账户导航树
     * 账户导航tree
     *
     * @param navigationVOMap 导航vomap
     * @return {@link List}<{@link AccountNavigationVO}>
     */
    private List<AccountNavigationVO> accountNavigationTree(Map<String, AccountNavigationVO> navigationVOMap) {
        // 查询所有标签
        List<WarningLabelTree> labelTreeList = warningLabelTreeServiceImpl.list();
        List<AccountNavigationVO> navigationVOList = labelTreeList.stream().map(warningLabelTree -> {
            AccountNavigationVO navigationVO = navigationVOMap.getOrDefault(warningLabelTree.getLabelValue(), new AccountNavigationVO());
            return AccountNavigationVO.build(warningLabelTree.getId(), warningLabelTree.getParentId(), warningLabelTree, navigationVO.getAccountCount(), navigationVO.getMessageCount());
        }).collect(Collectors.toList());
        List<AccountNavigationVO> parentNavigation = navigationVOList.stream().filter(accountNavigationVO -> Objects.equals(accountNavigationVO.getParentId(), 0L)).collect(Collectors.toList());
        Map<Long, List<AccountNavigationVO>> childrenNavigationGroup = navigationVOList.stream().filter(accountNavigationVO -> !Objects.equals(accountNavigationVO.getParentId(), 0L)).collect(Collectors.groupingBy(AccountNavigationVO::getParentId));
        childrenNavigationGroup.forEach((aLong, accountNavigationVOS1) -> accountNavigationVOS1.forEach(accountNavigationVO -> accountNavigationVO.setChildren(childrenNavigationGroup.getOrDefault(accountNavigationVO.getId(), Collections.emptyList()))));
        parentNavigation.forEach(accountNavigationVO -> accountNavigationVO.setChildren(childrenNavigationGroup.getOrDefault(accountNavigationVO.getId(), Collections.emptyList())));
        return parentNavigation;
    }


    /**
     * 添加警告帐户
     *
     * @param warningAccount 警告帐户
     */
    @Override
    @Transactional(rollbackFor = Exception.class)
    public void addWarningAccount(WarningAccount warningAccount) throws InterruptedException {
        // 判断账号是否存在
        WarningAccount joinAnalysis = warningAccountMapper.judgmentAccountAlreadyExists(warningAccount);
        AssertUtils.isNull(joinAnalysis, AccountExceptionEnum.ACCOUNT_EXIST);
        // 判断是否已经加入分析
        this.doAddAccount(warningAccount);

    }


    /**
     * 添加账户
     *
     * @param warningAccount 警告帐户
     */
    private void doAddAccount(WarningAccount warningAccount) throws InterruptedException {
        RLock lock = redissonClient.getLock("warning:account:add:" + warningAccount.getAccount() + warningAccount.getPlatform() + warningAccount.getAccountLink());
        try {
            if (lock.tryLock(6, TimeUnit.SECONDS)) {
                int insertResult = warningAccountMapper.insert(warningAccount);
                AssertUtils.isTrue(insertResult > 0, CommonEnum.FAIL);
                // 保存账号标签关系信息
                this.addAccountLabelRelation(warningAccount.getAccountLabel(), warningAccount.getId());
                // 查询是否已经加入分析此账号
                WarningAccount accountJoinAnalysis = warningAccountMapper.judgmentAccountJoinAnalysis(warningAccount, RequestHolderUtil.getUserId());
                // 未加入分析 保存用户和账号的关系
                if (accountJoinAnalysis == null) {
                    WarningAccountAnalysis warningAccountAnalysis = new WarningAccountAnalysis();
                    warningAccountAnalysis.setAnalysisUserId(RequestHolderUtil.getUserId());
                    warningAccountAnalysis.setAnalysisAccountId(warningAccount.getId());
                    warningAccountAnalysisServiceImpl.addAccountAnalysis(warningAccountAnalysis);
                }
            }
        } finally {
            if (lock.isHeldByCurrentThread()) {
                lock.unlock();
            }
        }
    }


    /**
     * 添加帐户标签关系
     *
     * @param label     标签
     * @param accountId 帐户id
     */
    private void addAccountLabelRelation(String label, Long accountId) {
        if (StringUtils.isNotEmpty(label)) {
            List<WarningAccountLabelRelation> labelList = Stream.of(label.split(StringPool.COMMA)).map(labelValue -> new WarningAccountLabelRelation(null, accountId, labelValue, null)).collect(Collectors.toList());
            warningAccountLabelServiceImpl.addBatchAccountLabel(labelList);
        }
    }


    /**
     * 加入列表分析账户
     *
     * @return {@link ListResponse}<{@link JoinAnalysisAccountVO}>
     */
    @Override
    public PageResponse<JoinAnalysisAccountVO> listJoinAnalysisAccount(AccountJoinedListRequest accountJoinedListRequest) {
        long countJoinAnalysisAccount = warningAccountMapper.countJoinAnalysisAccount(RequestHolderUtil.getUserId(), accountJoinedListRequest);
        if (countJoinAnalysisAccount == 0) {
            return PageResponse.builderEmpty();
        }
        List<WarningAccount> warningAccounts = warningAccountMapper.listJoinAnalysisAccount(RequestHolderUtil.getUserId(), accountJoinedListRequest);
        DictResponse dictResponse = dictServiceBlockingStub.dictList(DictRequest.newBuilder().setDictType("warning_message_source").build());
        Map<String, String> platformMap = dictResponse.getListList().stream().collect(Collectors.toMap(DictResponseData::getDictDataValue, DictResponseData::getDictDataName));
        List<Long> accountIds = warningAccounts.stream().map(WarningAccount::getId).collect(Collectors.toList());
        // 查询账号的标签
        List<WarningAccountLabelRelation> warningAccountLabelRelations = warningLabelTreeServiceImpl.queryLabelByAccount(accountIds);
        Map<Long, List<WarningAccountLabelRelation>> accountLabels = warningAccountLabelRelations.stream().collect(Collectors.groupingBy(WarningAccountLabelRelation::getWarningAccountId));
        List<JoinAnalysisAccountVO> accountVOList = warningAccounts.stream().map(warningAccount -> JoinAnalysisAccountVO.build(warningAccount, platformMap, accountLabels)).collect(Collectors.toList());
        return PageResponse.builder(accountVOList, countJoinAnalysisAccount);
    }


    /**
     * 预警账号列表
     *
     * @param warningAccountRequest 警告帐户请求
     * @return {@link PageResponse}<{@link AccountAlertDataVO}>
     */
    @Override
    public PageResponse<AccountAlertDataVO> listAccountAlert(WarningAccountRequest warningAccountRequest) {
        // 根据筛选条件查询预警账号
        Message message = ProtoJsonUtil.toProtoMessage(SearchAccountGrpcRequest.newBuilder(), JSON.toJSONString(warningAccountRequest));
        SearchAccountPageResponse searchAccountPageResponse = searchAccountServiceBlockingStub.searchLabelAccount((SearchAccountGrpcRequest) message);
        List<SearchAccountResponseData> responseDataList = searchAccountPageResponse.getListList();
        List<WarningAccountCollect> warningAccountCollects = new ArrayList<>();
        List<WarningAccountAnalysisParam> warningAccountAnalysisParams = new ArrayList<>();
        Map<String, String> platformNameMap = dictGrpcClient.getDictMap("warning_message_source");
        // 实体转换
        List<AccountAlertDataVO> alertDataVOList = responseDataList.stream().map(searchAccountResponseData -> {
            AccountAlertDataVO accountAlertDataVO = ProtoJsonUtil.toJava(searchAccountResponseData, AccountAlertDataVO.class);
            accountAlertDataVO.setPlatform(platformNameMap.get(searchAccountResponseData.getUserType()));
            accountAlertDataVO.setAvatar(searchAccountResponseData.getHeadImage());
            return accountAlertDataVO;
        }).peek(accountAlertDataVO -> {
            // 获取账号、查询这些账号是否在账号收藏和分析表中存在、判断是否已经收藏和加入分析
            warningAccountCollects.add(WarningAccountCollect.build(accountAlertDataVO.getUserName(), accountAlertDataVO.getPlatform(), RequestHolderUtil.getUserId()));
            warningAccountAnalysisParams.add(WarningAccountAnalysisParam.builder().analysisAccount(accountAlertDataVO.getUserName()).analysisUserId(RequestHolderUtil.getUserId()).platform(accountAlertDataVO.getPlatform()).build());
        }).collect(Collectors.toList());
        // 查询加入分析表 判断账号是否被加入分析
        List<WarningAccountCollect> batchQueryByParamList = warningAccountCollectServiceImpl.batchQueryByParamList(warningAccountCollects);
        Map<String, WarningAccountCollect> accountCollectMap = batchQueryByParamList.stream().collect(Collectors.toMap(warningAccountCollect -> warningAccountCollect.getCollectAccountId() + StringPool.DASH + warningAccountCollect.getCollectUserId() + StringPool.DASH + warningAccountCollect.getCollectAccountPlatform(), Function.identity()));
        // 账号加入分析状态
        List<WarningAccountAnalysisPO> warningAccountAnalysisPOS = warningAccountAnalysisServiceImpl.batchQueryAnalysisAccount(warningAccountAnalysisParams);
        Map<String, WarningAccountAnalysisPO> warningAccountAnalysisPOMap = warningAccountAnalysisPOS.stream().collect(Collectors.toMap(warningAccountAnalysisPO -> warningAccountAnalysisPO.getAccount() + StringPool.DASH + warningAccountAnalysisPO.getAnalysisUserId() + StringPool.DASH + warningAccountAnalysisPO.getPlatform(), Function.identity()));
        alertDataVOList.forEach(accountAlertDataVO -> {
            boolean isCollected = accountCollectMap.containsKey(accountAlertDataVO.getUserName() + StringPool.DASH + RequestHolderUtil.getUserId() + StringPool.DASH + accountAlertDataVO.getPlatform());
            // 收藏状态
            accountAlertDataVO.setCollectStatus(isCollected ? JOINED : NOT_JOINED);
            // 分析状态
            boolean isAnalysis = StringUtils.isNotEmpty(accountAlertDataVO.getUserName()) && warningAccountAnalysisPOMap.containsKey(accountAlertDataVO.getUserName() + StringPool.DASH + RequestHolderUtil.getUserId() + StringPool.DASH + accountAlertDataVO.getPlatform());
            accountAlertDataVO.setJoinedStatus(isAnalysis ? JOINED : NOT_JOINED);
        });
        return PageResponse.builder(alertDataVOList, searchAccountPageResponse.getTotal());
    }

    /**
     * 更新账号信息
     *
     * @param warningAccount 警告帐户
     */
    @Override
    public void updateAccountInfo(WarningAccount warningAccount) {
        AssertUtils.isNotNull(warningAccountMapper.selectById(warningAccount.getId()), AccountExceptionEnum.ACCOUNT_DOES_NOT_EXIST);
        int updateResult = warningAccountMapper.updateById(warningAccount);
        AssertUtils.isTrue(updateResult > 0, CommonEnum.FAIL);
    }

    /**
     * 获取账户基础信息
     *
     * @param id id
     * @return {@link WarningAccountVO}
     */
    @Override
    public WarningAccountVO getAccountBaseInfo(Long id) {
        WarningAccount warningAccount = warningAccountMapper.selectById(id);
        AssertUtils.isNotNull(warningAccount, AccountExceptionEnum.ACCOUNT_DOES_NOT_EXIST);
        return JSON.parseObject(JSON.toJSONString(warningAccount), WarningAccountVO.class);
    }

    /**
     * 删除账户
     *
     * @param id id
     */
    @Override
    @Transactional(rollbackFor = Exception.class)
    public void deleteAccount(Long id) {
        WarningAccount warningAccount = warningAccountMapper.selectById(id);
        AssertUtils.isNotNull(warningAccount, AccountExceptionEnum.ACCOUNT_DOES_NOT_EXIST);
        int deleteResult = warningAccountMapper.deleteById(warningAccount);
        AssertUtils.isTrue(deleteResult > 0, CommonEnum.FAIL);
        // 删除图
        //warningAccountPushGrpcClient.deleteGraph(KeyAccount.newBuilder().setId(id).build());
    }


    /**
     * 添加分析
     *
     * @param accountId 帐户id
     * @param platform  平台
     * @return {@link Long}
     */
    @Override
    public Long addAnalysis(String accountId, String platform) {
        WarningAccount warningAccount = new WarningAccount();
        warningAccount.setAccountId(accountId);
        warningAccount.setPlatform(platform);
        //查询账号的基本信息
        WarningAccountRequest warningAccountRequest = new WarningAccountRequest();
        warningAccountRequest.setPlatform(platform);
        warningAccountRequest.setSearchKeywords(accountId);
        PageResponse<AccountAlertDataVO> response = listAccountAlert(warningAccountRequest);
        if (response != null && !CollectionUtils.isEmpty(response.getList())) {
            Optional<AccountAlertDataVO> first = response.getList().stream().filter(o -> o.getId().equalsIgnoreCase(accountId) && o.getPlatform().equalsIgnoreCase(platform)).findFirst();
            if (first.isPresent()) {
                AccountAlertDataVO vo = first.get();
                warningAccount.setAccountAvatar(vo.getAvatar());
                warningAccount.setAccount(vo.getUserName());
                warningAccount.setAccountLabel(vo.getTags());
                warningAccount.setAccountLink(vo.getUrls());
                warningAccount.setAccountNickName(vo.getNickName());
                warningAccount.setAccountSummary(vo.getDescription());
            }
        }
        try {
            addWarningAccount(warningAccount);
        } catch (InterruptedException e) {
            log.error("线程被中断， 异常详情：{}", e);
        }

        //通知采集任务
        try {
            ApplicationContextUtil.getBean(ExecutorService.class, "defaultThreadPoolExecutor")
                    .execute(() -> warningDataSyncNotifyService.notify(accountId, platform));
        } catch (Exception e) {
            log.error("加入分析通知采集时出现异常， 异常详情：{}", e);
        }

        return warningAccount.getId();
    }

}




